package com.qhx.rtkcloud.ntrip;


import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketOptions;
import java.util.*;
import java.util.concurrent.*;

//@Configuration
public class NtripCaster {

    public static Map<String, Socket> serverMap = new HashMap();
    public static Map<String, Map<Socket, LinkedBlockingQueue<byte[]>>> mapQueue = new ConcurrentHashMap<>();
    public static ExecutorService executorService = null;


    @Value("${rtk.socketPort}")
    private Integer socketPort;


    @Bean(destroyMethod = "close")
    public ServerSocket serverSocket() throws IOException {
        executorService = Executors.newFixedThreadPool(20);
        return new ServerSocket(socketPort); // 使用2101端口
    }


    @Bean
    public SocketListener socketListener(ServerSocket serverSocket) {
        return new SocketListener(serverSocket);
    }

    private void registerMount(Socket clientSocket , String commandLine) throws IOException {
        String mount = commandLine.substring(commandLine.lastIndexOf("/") + 1).trim();
        InputStream socketInputStream = clientSocket.getInputStream();
        InputStreamReader inputStreamReader = new InputStreamReader(socketInputStream);
        BufferedReader in = new BufferedReader(inputStreamReader);
        PrintWriter out = new PrintWriter(clientSocket.getOutputStream());
        String os = System.getProperty("os.name");
        if (os.toLowerCase().contains("windows")) {
            out.println("ICY 200 OK");
        } else if (os.toLowerCase().contains("linux")) {
            //同一个基准站有可能发多次连接请求
            out.println("ICY 200 OK\r");
        }
        out.flush();
        //将基准站对应的挂载点添加到serverMap,serverMap中有挂载点的数据证明已经连接
        serverMap.put(mount, clientSocket);
        //rtkConnectBreakSocketService
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    sendMessage(clientSocket, mount);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }


    class SocketListener {
        private final ServerSocket serverSocket;

        public SocketListener(ServerSocket serverSocket) {
            this.serverSocket = serverSocket;
            start();
        }

        public void start() {
            new Thread(() -> {
                while (true) {
                    try {
                        //监听2101端口等待基准站/本地服务连接
                        Socket clientSocket = serverSocket.accept();

                      //  clientSocket.setSoTimeout(10000);
                      //  clientSocket.setKeepAlive(true);
                        // 与基准站建立连接后，开始处理Ntrip协议的交互
                        InputStream socketInputStream = clientSocket.getInputStream();

                        InputStreamReader inputStreamReader = new InputStreamReader(socketInputStream);
                        BufferedReader in = new BufferedReader(inputStreamReader);
                        PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);
                        String inputLine;
                        Socket serverSocket = null;
                        //云端基准站发送的数据
                        if ((inputLine = in.readLine()) != null) {
                            if (inputLine.startsWith("SOURCE")) {
                                String mount = inputLine.substring(inputLine.lastIndexOf("/") + 1).trim();
                                String os = System.getProperty("os.name");
                                if (os.toLowerCase().contains("windows")) {
                                    out.println("ICY 200 OK");
                                } else if (os.toLowerCase().contains("linux")) {
                                    //同一个基准站有可能发多次连接请求
                                    out.println("ICY 200 OK\r");
                                }
                                out.flush();
                                //将基准站对应的挂载点添加到serverMap,serverMap中有挂载点的数据证明已经连接
                                serverMap.put(mount, clientSocket);
                                //rtkConnectBreakSocketService
                                executorService.submit(new Runnable() {
                                    @Override
                                    public void run() {
                                        try {
                                            sendMessage(clientSocket, mount);
                                        } catch (Exception e) {
                                            try {
                                                clientSocket.close();
                                            } catch (IOException ex) {
                                                ex.printStackTrace();
                                            }
                                            serverMap.remove(mount);
                                        }
                                    }
                                });
                                continue;
                            } else if (inputLine.startsWith("GET /")) {//本地服务发送的连接请求,连接成功从相应的mapQueue中获取数据相应的数据
                                String mount = null;
                                if (inputLine.contains(" HTTP/1.1")) {
                                    mount = inputLine.substring(inputLine.indexOf("GET /") + 5, inputLine.indexOf(" HTTP/1.1")).trim();
                                } else if (inputLine.contains(" HTTP/1.0")) {
                                    mount = inputLine.substring(inputLine.indexOf("GET /") + 5, inputLine.indexOf(" HTTP/1.0")).trim();
                                }
                                out.println("ICY 200 OK");
                                out.flush();
                                //查看是否已经有相应的基准站连接到到云端服务,如果没有直接返回,等待下次重连
                                serverSocket = serverMap.get(mount);
                                if (serverSocket == null) {
                                    out.println("HTTP/1.0 401 Unauthorized");
                                    continue;
                                }

                                Map<Socket, LinkedBlockingQueue<byte[]>> socketQueueMap = mapQueue.get(mount);
                                if (socketQueueMap != null) {
                                    //这一步的目的是查看是否有重复的连接,如果有证明之前的连接失效,删除之前的连接改用新连接
                                    Set<Socket> sockets = socketQueueMap.keySet();
                                    String ip = clientSocket.getInetAddress().getHostAddress();
                                    Iterator<Socket> iterator = sockets.iterator();
                                    while (iterator.hasNext()) {
                                        Socket socket = iterator.next();
                                        if (ip.equals(socket.getInetAddress().getHostAddress())) {
                                            iterator.remove();
                                        }
                                    }
                                    LinkedBlockingQueue<byte[]> queue = new LinkedBlockingQueue();
                                    socketQueueMap.put(clientSocket, queue);
                                } else {
                                    Map map = new HashMap();
                                    BlockingQueue<byte[]> queue = new LinkedBlockingQueue();
                                    map.put(clientSocket, queue);
                                    mapQueue.put(mount, map);
                                }

                                getMessage(mount, clientSocket);
                            }
                        }
                        continue;
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }


    public void sendMessage(Socket serverSocket, String mount) throws IOException, InterruptedException {
        byte[] buffer = new byte[10240];

        InputStream serverSocketInputStream = serverSocket.getInputStream();
        while (true) {
            int bytesRead = 0;
            try {
                //serverSocket.sendUrgentData(0xff);
                bytesRead = serverSocketInputStream.read(buffer);
            } catch (Exception e) {
                serverSocket.close();
                serverMap.remove(mount);
                //如果基站链接断开,那相应的本地client链接都需要断开
                Map<Socket, LinkedBlockingQueue<byte[]>> socketQueueMap = mapQueue.get(mount);
                if (socketQueueMap != null) {
                    for (Socket socket : socketQueueMap.keySet()) {
                        socket.close();
                    }
                }
                e.printStackTrace();
                break;
            }
            Map<Socket, LinkedBlockingQueue<byte[]>> socketQueueMap = mapQueue.get(mount);
            if (socketQueueMap != null && socketQueueMap.size() > 0) {
                for (Socket socket : socketQueueMap.keySet()) {
                    Queue<byte[]> socketQueue = socketQueueMap.get(socket);
                    if (bytesRead > 0) {
                        byte[] bytes = Arrays.copyOfRange(buffer, 0, bytesRead);
                        socketQueue.offer(bytes);
                        buffer = new byte[10240];
                    }
                }
            } else {
                buffer = new byte[10240];
                Thread.sleep(1000);
            }
        }

    }


    public void getMessage(String mount, Socket clientSocket) throws IOException {


        executorService.submit(new Runnable() {
            @Override
            public void run() {
                OutputStream outputStream = null;
                try {
                    outputStream = clientSocket.getOutputStream();
                    while (true) {
                        Map<Socket, LinkedBlockingQueue<byte[]>> queueMap = mapQueue.get(mount);
                        if (queueMap.keySet().size() > 0) {
                            LinkedBlockingQueue<byte[]> queue1 = queueMap.get(clientSocket);
                            byte[] poll = queue1.take();
                            outputStream.write(poll);
                            outputStream.flush();
                            System.out.println("发送给客户端的数据"+mount + Arrays.toString(poll));
                        }else {
                            Thread.sleep(2000);
                        }
                    }
                } catch (Exception e) {
                    Map<Socket, LinkedBlockingQueue<byte[]>> queueMap = mapQueue.get(mount);
                    for (Socket socket : queueMap.keySet()) {
                        try {
                            socket.close();
                        } catch (IOException ex) {
                            throw new RuntimeException(ex);
                        }
                    }
                    if (queueMap != null) {
                        mapQueue.remove(mount);
                    }
                }
            }
        });
    }
}
